/*
 * Copyright 1999-2011 Alibaba Group.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.alibaba.dubbo.rpc.cluster.loadbalance;

import com.alibaba.dubbo.common.URL;
import com.alibaba.dubbo.common.utils.AtomicPositiveInteger;
import com.alibaba.dubbo.rpc.Invocation;
import com.alibaba.dubbo.rpc.Invoker;

import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 * 加权轮循
 *
 *  最大权重 为 0
 *  最小权重 为 Integer.MAX_VALUE
 *  【平滑加权轮询算法】
 *
 * @author qian.lei
 * @author william.liangf
 */
public class RoundRobinLoadBalance extends AbstractLoadBalance {

    public static final String NAME = "roundrobin";

    private final ConcurrentMap<String, AtomicPositiveInteger> sequences = new ConcurrentHashMap<String, AtomicPositiveInteger>();

    /**
     * 处理逻辑:
     * 1. 检查候选者权重值是否相同（比较最大、最小值是否相等）、计算【总权重值】。
     *    按照顺序组织好候选者与权重值的对应关系（LinkedHashMap）
     * 2. 从缓存中取出请求方法的次序号+1，得到当前请求的次序号
     *  （区分请求的粒度为接口的方法级别、如果没有缓存则初始化缓存，序号为0，+1后当前次序号为1）
     * 3. 如果候选者权重相同: 那么直接轮询就好了，用【次序号】对【候选者个数】取模，【余数】对应下标的invoker即为候选者
     *    如果候选者权重不同: 用【次序号】对【总权重值】取模得到【mod】,
     *                     用最大的候选者权重做循环次数, 轮流依次对每个候选者的权重值-1, 同时 mod--, 如果权重值比较小的权重就会先被减到0, 减到0的直接忽略,
     *                     直到mod被减到0, 并且候选者的权重还有剩余, 那么这个候选者就是被选中的。
     *                     换一种解释方式:
     *                         权重总数等于机会数，总机会数用光了就重置游戏。（【次序号】对【总权重值】取模不会超过【总权重值】）
     *                         下面4个候选者，权重依次是 甲:1  乙:4  丙:2  丁:3,
     *                         也就是甲有1次选中机会、乙有4次机会、丙丁同理,玩游戏就需要公平，按照 甲->乙->丙->丁-> 甲->乙->丙->...
     *                         谁机会用光了就出局等待下次重置游戏。
     *                         如下表格: 纵向为循环次数、横向为候选者被选中的次序，这个次序值就是【次序号】对【总权重值】取模的值
     *                              -------------------------
     *                              循环次数  | 1 | 2 | 3 | 4 |    循环次数为候选者最大权重值，只有它的机会用光了才能重置游戏
     *                              -------------------------
     *                              下  | 甲 | 0 |   |   |   |    甲 1 次机会
     *                                  ---------------------
     *                              标  | 乙 | 1 | 4 | 7 | 9 |    乙 4 次机会
     *                                  ---------------------
     *                              数  | 丙 | 2 | 5 |   |   |    丙 2 次机会
     *                                  ---------------------
     *                              字  | 丁 | 3 | 6 | 8 |   |    丁 3 次机会
     *                              -------------------------
     *
     * 可以看出算法的复杂度受 mod 和 大权重值 影响，如果候选者配置了很大的权重值，并且mod也很大时，循环次数会很大，性能严重下降
     * 新版本对此代码做了修改
     * 参见：http://dubbo.apache.org/zh-cn/docs/source_code_guide/loadbalance.html
     * 新版代码贴在此类的最下方作为参考。
     *
     * @param invokers
     * @param url
     * @param invocation
     * @param <T>
     * @return
     */
    @Override
    protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
        String key = invokers.get(0).getUrl().getServiceKey() + "." + invocation.getMethodName();
        int length = invokers.size(); // 总个数
        int maxWeight = 0; // 最大权重
        int minWeight = Integer.MAX_VALUE; // 最小权重
        final LinkedHashMap<Invoker<T>, IntegerWrapper> invokerToWeightMap = new LinkedHashMap<Invoker<T>, IntegerWrapper>();
        int weightSum = 0;
        for (int i = 0; i < length; i++) {
            int weight = getWeight(invokers.get(i), invocation);
            maxWeight = Math.max(maxWeight, weight); // 寻找最大权重
            minWeight = Math.min(minWeight, weight); // 寻找最小权重
            if (weight > 0) {
                invokerToWeightMap.put(invokers.get(i), new IntegerWrapper(weight));
                weightSum += weight; // 累计权重和
            }
        }
        AtomicPositiveInteger sequence = sequences.get(key);
        if (sequence == null) {
            sequences.putIfAbsent(key, new AtomicPositiveInteger());
            sequence = sequences.get(key);
        }
        int currentSequence = sequence.getAndIncrement();
        if (maxWeight > 0 && minWeight < maxWeight) { // 权重不一样
            int mod = currentSequence % weightSum;
            for (int i = 0; i < maxWeight; i++) {
                for (Map.Entry<Invoker<T>, IntegerWrapper> each : invokerToWeightMap.entrySet()) {
                    final Invoker<T> k = each.getKey();
                    final IntegerWrapper v = each.getValue();
                    if (mod == 0 && v.getValue() > 0) {
                        return k;
                    }
                    if (v.getValue() > 0) {
                        v.decrement();
                        mod--;
                    }
                }
            }
        }
        // 取模轮循
        return invokers.get(currentSequence % length);
    }

    private static final class IntegerWrapper {
        private int value;

        public IntegerWrapper(int value) {
            this.value = value;
        }

        public int getValue() {
            return value;
        }

        public void setValue(int value) {
            this.value = value;
        }

        public void decrement() {
            this.value--;
        }
    }

/////////////////////////////////////////////// 新版代码优化版本 ////////////////////////////////////////////////////////////////////////////////////
//
//public class RoundRobinLoadBalance extends AbstractLoadBalance {
//    public static final String NAME = "roundrobin";
//
//    private static int RECYCLE_PERIOD = 60000;
//
//    protected static class WeightedRoundRobin {
//        // 服务提供者权重
//        private int weight;
//        // 当前权重
//        private AtomicLong current = new AtomicLong(0);
//        // 最后一次更新时间
//        private long lastUpdate;
//
//        public void setWeight(int weight) {
//            this.weight = weight;
//            // 初始情况下，current = 0
//            current.set(0);
//        }
//        public long increaseCurrent() {
//            // current = current + weight；
//            return current.addAndGet(weight);
//        }
//        public void sel(int total) {
//            // current = current - total;
//            current.addAndGet(-1 * total);
//        }
//    }
//
//    // 嵌套 Map 结构，存储的数据结构示例如下：
//    // {
//    //     "UserService.query": {
//    //         "url1": WeightedRoundRobin@123,
//    //         "url2": WeightedRoundRobin@456,
//    //     },
//    //     "UserService.update": {
//    //         "url1": WeightedRoundRobin@123,
//    //         "url2": WeightedRoundRobin@456,
//    //     }
//    // }
//    // 最外层为服务类名 + 方法名，第二层为 url 到 WeightedRoundRobin 的映射关系。
//    // 这里我们可以将 url 看成是服务提供者的 id
//    private ConcurrentMap<String, ConcurrentMap<String, WeightedRoundRobin>> methodWeightMap =
//          new ConcurrentHashMap<String, ConcurrentMap<String, WeightedRoundRobin>>();
//
//    // 原子更新锁
//    private AtomicBoolean updateLock = new AtomicBoolean();
//
//    @Override
//    protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
//        String key = invokers.get(0).getUrl().getServiceKey() + "." + invocation.getMethodName();
//        // 获取 url 到 WeightedRoundRobin 映射表，如果为空，则创建一个新的
//        ConcurrentMap<String, WeightedRoundRobin> map = methodWeightMap.get(key);
//        if (map == null) {
//            methodWeightMap.putIfAbsent(key, new ConcurrentHashMap<String, WeightedRoundRobin>());
//            map = methodWeightMap.get(key);
//        }
//        int totalWeight = 0;
//        long maxCurrent = Long.MIN_VALUE;
//
//        // 获取当前时间
//        long now = System.currentTimeMillis();
//        Invoker<T> selectedInvoker = null;
//        WeightedRoundRobin selectedWRR = null;
//
//        // 下面这个循环主要做了这样几件事情：
//        //   1. 遍历 Invoker 列表，检测当前 Invoker 是否有
//        //      相应的 WeightedRoundRobin，没有则创建
//        //   2. 检测 Invoker 权重是否发生了变化，若变化了，
//        //      则更新 WeightedRoundRobin 的 weight 字段
//        //   3. 让 current 字段加上自身权重，等价于 current += weight
//        //   4. 设置 lastUpdate 字段，即 lastUpdate = now
//        //   5. 寻找具有最大 current 的 Invoker，以及 Invoker 对应的 WeightedRoundRobin，
//        //      暂存起来，留作后用
//        //   6. 计算权重总和
//        for (Invoker<T> invoker : invokers) {
//            String identifyString = invoker.getUrl().toIdentityString();
//            WeightedRoundRobin weightedRoundRobin = map.get(identifyString);
//            int weight = getWeight(invoker, invocation);
//            if (weight < 0) {
//                weight = 0;
//            }
//
//            // 检测当前 Invoker 是否有对应的 WeightedRoundRobin，没有则创建
//            if (weightedRoundRobin == null) {
//                weightedRoundRobin = new WeightedRoundRobin();
//                // 设置 Invoker 权重
//                weightedRoundRobin.setWeight(weight);
//                // 存储 url 唯一标识 identifyString 到 weightedRoundRobin 的映射关系
//                map.putIfAbsent(identifyString, weightedRoundRobin);
//                weightedRoundRobin = map.get(identifyString);
//            }
//            // Invoker 权重不等于 WeightedRoundRobin 中保存的权重，说明权重变化了，此时进行更新
//            if (weight != weightedRoundRobin.getWeight()) {
//                weightedRoundRobin.setWeight(weight);
//            }
//
//            // 让 current 加上自身权重，等价于 current += weight
//            long cur = weightedRoundRobin.increaseCurrent();
//            // 设置 lastUpdate，表示近期更新过
//            weightedRoundRobin.setLastUpdate(now);
//            // 找出最大的 current
//            if (cur > maxCurrent) {
//                maxCurrent = cur;
//                // 将具有最大 current 权重的 Invoker 赋值给 selectedInvoker
//                selectedInvoker = invoker;
//                // 将 Invoker 对应的 weightedRoundRobin 赋值给 selectedWRR，留作后用
//                selectedWRR = weightedRoundRobin;
//            }
//
//            // 计算权重总和
//            totalWeight += weight;
//        }
//
//        // 对 <identifyString, WeightedRoundRobin> 进行检查，过滤掉长时间未被更新的节点。
//        // 该节点可能挂了，invokers 中不包含该节点，所以该节点的 lastUpdate 长时间无法被更新。
//        // 若未更新时长超过阈值后，就会被移除掉，默认阈值为60秒。
//        if (!updateLock.get() && invokers.size() != map.size()) {
//            if (updateLock.compareAndSet(false, true)) {
//                try {
//                    ConcurrentMap<String, WeightedRoundRobin> newMap = new ConcurrentHashMap<String, WeightedRoundRobin>();
//                    // 拷贝
//                    newMap.putAll(map);
//
//                    // 遍历修改，即移除过期记录
//                    Iterator<Entry<String, WeightedRoundRobin>> it = newMap.entrySet().iterator();
//                    while (it.hasNext()) {
//                        Entry<String, WeightedRoundRobin> item = it.next();
//                        if (now - item.getValue().getLastUpdate() > RECYCLE_PERIOD) {
//                            it.remove();
//                        }
//                    }
//
//                    // 更新引用
//                    methodWeightMap.put(key, newMap);
//                } finally {
//                    updateLock.set(false);
//                }
//            }
//        }
//
//        if (selectedInvoker != null) {
//            // 让 current 减去权重总和，等价于 current -= totalWeight
//            selectedWRR.sel(totalWeight);
//            // 返回具有最大 current 的 Invoker
//            return selectedInvoker;
//        }
//
//        // should not happen here
//        return invokers.get(0);
//    }
//}
//
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////

}